package io.kiki.sba.registry.store.jdbc.informer;


import io.kiki.sba.registry.store.api.meta.DbEntry;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import lombok.Getter;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public abstract class BaseInformer<T extends DbEntry, C extends DbEntryContainer<T>> {
    private static final Logger logger = LoggerFactory.getLogger(BaseInformer.class);

    private static final int DB_INSERT_DELAY_MS = 1000;
    protected final Watch watchLoop = new Watch();
    protected final List listLoop = new List();
    private final Lock listLock = new ReentrantLock();
    private final String name;

    @Getter
    protected volatile C container = containerFactory();
    protected int watchLoopIntervalMs = 1000;
    protected int listLoopIntervalMs = 1000 * 60 * 30;
    @Getter
    private volatile long lastLoadId;
    private volatile boolean enabled;
    private volatile boolean listLoopGotLock;
    private boolean started;
    private volatile long syncStartVersion;
    private volatile long syncEndVersion;
    private volatile boolean allSynced = false;

    public BaseInformer(String name) {
        this.name = name;

    }


    public synchronized void start() {
        if (started) {
            return;
        }
        ConcurrentUtils.createDaemonThread(name + "-WatchLoop", watchLoop).start();
        ConcurrentUtils.createDaemonThread(name + "-ListLoop", listLoop).start();
        started = true;
        logger.info("{}-Informer started", name);
    }

    private void watch() {
        syncStart();
        try {
            long start = lastLoadId;
            if (listStableEntries(start, 1).size() == 0) {
                return;
            }
            logger.info("start watch from {}", start);
            long maxId = listToTail((T entry) -> {
                container.onEntry(entry);
                logger.info("watch received entry: {}", entry);
            }, start, 100);
            logger.info("end watch to {}", maxId);
            lastLoadId = maxId;
        } finally {
            syncEnd();
        }
    }

    private void list() {
        syncStart();
        try {
            C newContainer = containerFactory();
            long maxId = listToTail(newContainer::onEntry, 0, 1000);
            logger.info("end list to {}", maxId);
            preList(newContainer);
            this.container = newContainer;
            lastLoadId = maxId;
        } finally {
            syncEnd();
        }
    }

    private long listToTail(EntryCallable<T> callable, final long start, final int page) {
        long curStart = start;
        while (true) {
            java.util.List<T> entries = listStableEntries(curStart, page);
            if (CollectionUtils.isEmpty(entries)) {
                break;
            }
            for (T entry : entries) {
                callable.onEntry(entry);
                curStart = Math.max(curStart, entry.getId());
            }
            ConcurrentUtils.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
        }
        return curStart;
    }

    private java.util.List<T> listStableEntries(long start, int limit) {
        Date now = getNow();
        java.util.List<T> entries = listFromStorage(start, limit);
        allSynced = false;
        if (CollectionUtils.isEmpty(entries)) {
            allSynced = true;
            return Collections.emptyList();
        }
        java.util.List<T> result = new ArrayList<>(entries.size());
        for (T entry : entries) {
            if (entry.getGmtCreate().getTime() >= now.getTime() - DB_INSERT_DELAY_MS) {
                break;
            }
            result.add(entry);
        }
        return result;
    }

    public void setEnabled(boolean enabled) {
        this.enabled = enabled;
        if (enabled) {
            listLoop.wakeup();
        }
    }

    public void listWakeup() {
        listLoop.wakeup();
    }

    private void syncStart() {
        syncStartVersion = System.currentTimeMillis();
    }

    private void syncEnd() {
        syncEndVersion = System.currentTimeMillis();
    }

    public void waitSynced() {
        long version = System.currentTimeMillis();
        while (true) {
            if (syncStartVersion > version && syncEndVersion > syncStartVersion && allSynced) {
                long end = System.currentTimeMillis();
                logger.info("{}-waitSynced, start:{}, end:{}, cost:{}", name, version, end, end - version);
                return;
            }
            ConcurrentUtils.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
        }
    }

    protected abstract C containerFactory();

    protected abstract java.util.List<T> listFromStorage(long start, int limit);

    protected abstract Date getNow();

    protected void preList(C newContainer) {
    }

    public void watchWakeup() {
        watchLoop.wakeup();
    }

    interface EntryCallable<T> {
        void onEntry(T entry);
    }

    private final class Watch extends WakeUpLoopExecuteTask {

        @Override
        public int getWaitingMillis() {
            return watchLoopIntervalMs;
        }

        @Override
        public void _execute_() {
            if (!enabled) {
                return;
            }
            // ensure startup period listWatch executed in front
            if (!listLoopGotLock) {
                return;
            }
            listLock.lock();
            try {
                watch();
            } finally {
                listLock.unlock();
            }
        }
    }

    private final class List extends WakeUpLoopExecuteTask {
        @Override
        public int getWaitingMillis() {
            int base = listLoopIntervalMs / 2;
            return (int) (base + Math.random() * base);
        }

        @Override
        public void _execute_() {
            if (!enabled) {
                return;
            }
            listLock.lock();
            listLoopGotLock = true;
            try {
                list();
            } finally {
                listLock.unlock();
            }
        }
    }
}
